package com.ucode.dcse;

import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
import com.ucode.tool.disruptor.DisruptorConsumer;
import com.ucode.tool.disruptor.DisruptorExceptionHandler;
import com.ucode.tool.disruptor.DisruptorHandleService;
import com.ucode.tool.disruptor.DisruptorProducer;
import com.ucode.tool.disruptor.DisruptorThreadFactory;

@Component("canalSyncEsDisruptorProducer")
public class CanalSyncEsDisruptorProducer extends DisruptorProducer implements DisposableBean, ApplicationListener<ContextRefreshedEvent>{

    private static String producerName = "Canal->ES:";
    
    private DisruptorHandleService disruptorHandleService;
    
    @Autowired
    public CanalSyncEsDisruptorProducer(@Qualifier("canalSyncEsDisruptorHandleService") DisruptorHandleService disruptorHandleService) {
        super(producerName, false);
        this.disruptorHandleService = disruptorHandleService;
    }

    @Override
    public void setHandleEvents() {
      //CPU可使用数量
      final Executor executor = new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), DisruptorThreadFactory
              .create(super.producerName+"Disruptor consumer-", false), new ThreadPoolExecutor.AbortPolicy());
      
      DisruptorConsumer[] disruptorConsumers = new DisruptorConsumer[threads];
      
      for (int i = 0; i < threads; i++){
          disruptorConsumers[i] = new DisruptorConsumer(disruptorHandleService, executor);
      }
      
      disruptor.handleEventsWithWorkerPool(disruptorConsumers);
        
    }

    @Override
    public void setDefaultExceptionHandler() {
      //设置异常处理
      disruptor.setDefaultExceptionHandler(new DisruptorExceptionHandler(disruptorHandleService));
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        super.start();        
    }

    @Override
    public void destroy() throws Exception {
        super.stop();
    }

}
